package com.rocketmq.producer.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.google.common.base.Objects;
import com.rocketmq.producer.config.MqConfig;
import com.rocketmq.producer.model.MseeageEntity;
import com.rocketmq.producer.service.MqService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * @Description:
 * @author: zhengfei
 * @param:
 * @date: 2018-04-16 下午2:12
 */
@Service
@Slf4j
public class MqServiceImpl implements MqService {
    @Autowired
    private MqConfig mqConfig;

    @Value("${testMqTopic}")
    private String testMqTopic;
    @Value("${testMqTopicTag}")
    private String testMqTopicTag;

    private DefaultMQProducer producer;

    private static final MessageQueueSelector SELECTOR = new MessageQueueSelector() {
        @Override
        public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
            int index;
            if (o == null) {
                index = 0;
            } else {
                index = o.hashCode() % list.size();
            }
            return list.get(index);
        }
    };

    @PostConstruct
    public void init() throws Exception {
        producer = new DefaultMQProducer("rocketmq-test");
        producer.setNamesrvAddr(mqConfig.getNamesrvAddr());
        producer.setPollNameServerInteval(10000);
        producer.start();
    }

    @Override
    public boolean sendMsq(MseeageEntity entity) {
        if (Objects.equal(entity, null) || StringUtils.isBlank(entity.getId())) {
            return false;
        }
        try {
            String shardingKey = entity.getId();
            String json = JSON.toJSONString(entity);
            Message msg = new Message(testMqTopic, testMqTopicTag, shardingKey, json.getBytes("UTF-8"));
            // 发送顺序消息
            SendResult sendResult = producer.send(msg, SELECTOR , Long.parseLong(shardingKey));
            if (Objects.equal(sendResult, null) || !Objects.equal(sendResult.getSendStatus(), SendStatus.SEND_OK)) {
                log.error("sendMsq fail, entity:{}", JSON.toJSONString(entity));
                return false;
            }
            log.info("sendMsq success, entity:{}", JSON.toJSONString(entity));
            return true;
        } catch (Exception e) {
            log.error("sendMsq fail, entity:{} ,e:{}", JSON.toJSONString(entity), e);
        }
        return false;
    }
}
